package com.athuahua.rector;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import reactor.util.context.Context;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author: huahua
 * @Date: 2024/11/11 0011
 * @Time: 15:25
 * @Description:
 */
public class APITest {
    /**
     * log()
     * onSubscribe 流被订阅
     * request(unbounded) 请求无限数据
     * onNext(N) 每个数据到达
     * onComplete() 流结束
     */
    @Test //数据过滤
    public void filterTest() {
        Flux.just(1, 2, 3, 4)
                //.log() 1,2,3,4
                .filter(ele -> ele % 2 == 0)
                .log() //2,4
                .subscribe();
    }

    @Test //数据包装成多元素流
    public void flatMap() {
        Flux.just("zhang san", "li si")
                .flatMap(v -> { //两个人的名字按照空格拆分，单独打印姓和名
                    String[] s = v.split(" ");
                    return Flux.fromArray(s);
                })
                .log()
                .subscribe();
    }

    @Test //连接映射 拼接流
    public void concatMap() {
        /**
         * concatMap 一个元素可以变成很多单个，对于元素类型无限制
         * concat: Flux.concat
         * concatWith 连接的流和老流的元素类型要一致
         */
        Flux.concat(Flux.just(1, 2), Flux.just(3, 4))
                .log()
                .subscribe();

        Flux.just(1, 2)
                .concatMap(s -> {
                    return Flux.just(s + "-a");
                })
                .log()
                .subscribe();
    }

    @Test //流转换
    public void transform() {
        /**
         * transform【无状态转换】：不会共享外部变量的值
         * transformDeferred【有状态转换】：共享外部变量的值
         */
        AtomicInteger atomic = new AtomicInteger(0);

        Flux<String> flux = Flux.just("a", "b", "c")
                .transform(values -> {
                    if (atomic.incrementAndGet() == 1) {
                        //atomic++ 再获取
                        //如果是第一次调用，老流的全部元素转成大写
                        return values.map(String::toUpperCase);
                    } else {
                        return values;
                    }
                });
        flux.subscribe(v -> System.out.println("订阅者1： v = " + v));
        flux.subscribe(v -> System.out.println("订阅者2： v = " + v));
    }

    @Test //按发布时间合并流
    public void merge() throws InterruptedException {
        /**
         * concat: 连接：A流和 B流拼接
         * merge: 合并：A流和 B流按照时间序列合并
         *
         * mergeWith
         * mergeSequential 按照那个流先发元素排队
         */
        Flux.merge(Flux.just(1, 2, 3).delayElements(Duration.ofSeconds(1)),
                Flux.just("a", "b").delayElements(Duration.ofMillis(1500)),
                Flux.just("haha", "hehe", "xixi").delayElements(Duration.ofMillis(500))
        ).log().subscribe();


        Thread.sleep(10000);

        //Flux.just(1, 2, 3).mergeWith(Flux.just(4, 5, 6));
    }


    @Test //压缩
    public void zip() {
        //zip期间无法结对的元素会被忽略,最多支持8流压缩
        // Tuple 元组
        Flux.just(1, 2, 3)
                .zipWith(Flux.just("a", "b", "c", "d"))
                .map(tuple -> {
                    return tuple.getT1() + "==>" + tuple.getT2();
                })
                .log()
                .subscribe();
    }


    @Test //空处理
    public void empty() {
        /**
         * defaultIfEmpty 兜底
         * swichIfEmpty 动态的兜底方法
         */

        haha()
                .defaultIfEmpty("x") //如果发布者元素为空，指定默认值
//                .switchIfEmpty(xixi())
                .subscribe(v -> System.out.println("v = " + v));
    }

    Mono<String> haha() {
        //Mono.just(null) 表示流中有一个null值元素
//        return Mono.just("a");
        return Mono.empty();
    }

    Mono<String> xixi() {
        return Mono.just("哈哈");
    }


    @Test //错误处理
    public void error() throws InterruptedException {
        /**
         * 默认：错误是一种中断行为
         * 命令式编程：常见的错误处理
         * 1.捕获异常，返回一个静态默认值
         * 2.吃掉异常，执行一个兜底方法
         * 3.捕获并动态计算一个返回值 【根据错误返回新值】
         * 4.捕获并保证成业务异常。并重新抛出 【常用】
         * 5.捕获异常，记录错误日志，并重新抛出
         * 6.无论错误和正常都执行的方法 finally
         * 7.忽略当前异常，通知记录，继续记录
         *
         *
         * 响应式编程
         * 1.onErrorReturn 返回默认值 [1.吃掉异常，消费者无异常感知 2.返回一个默认值 3.中断流,流正常完成]
         * 2.onErrorResume 返回兜底方法 [同上13 返回一个兜底方法]
         * 3.onErrorMap 捕获并保证业务异常并重新抛出 [消费者有异常感知 流异常完成 调用一个自定义方法]
         * 4.doOnError 感知异常，不会吃掉异常
         * 5.doFinally 一定会触发
         * 6.onErrorContinue 忽略异常，执行方法，继续执行
         * 7.onErrorComplete  错误结束信号替换成正常信号
         */
        Flux<Integer> map = Flux.just(1, 2, 0, 4)
                .map(i -> 100 / i);
        //1.onErrorReturn
        //map.onErrorReturn(666)
        //如果是数学运算异常，返回666
        //map.onErrorReturn(ArithmeticException.class,666)

        //2.onErrorResume
        //map.onErrorResume(err -> Mono.just(666))
        //map.onErrorResume(err -> hahaha(err))

        //3.onErrorMap
        //map.onErrorResume(err-> Flux.error(new BusinessException(err.getMessage()+ " 炸了")))
        //map.onErrorMap(err->new BusinessException(err.getMessage() + " 更简单的炸了。。。"))

        //4.doOnError
        map.doOnError(err -> {
                    System.out.println("err已被记录 = " + err);
                })
                //5. doFinally
                .doFinally(signalType -> System.out.println("流信号：" + signalType))

                .subscribe(
                        v -> System.out.println("消费流 = " + v), //消费流
                        err -> System.out.println("流错误处理 = " + err), //错误处理
                        () -> System.out.println("流结束") //流正常结束
                );
        System.out.println("-=--------------------------------");


        //6.onErrorContinue
        Flux.just(1, 2, 3, 0, 5)
                .map(i -> 10 / i)
                .onErrorContinue((err, val) -> {
                    System.out.println("err = " + err);
                    System.out.println("val = " + val);
                    System.out.println("发现" + val + "有问题，继续执行, 记录这个问题");
                })
                .subscribe(v -> System.out.println("v = " + v),
                        err -> System.out.println("err = " + err));

        System.out.println("-=--------------------------------");

        //7.onErrorComplete() 错误结束信号替换成正常信号
        Flux.interval(Duration.ofSeconds(1))
                .map(i -> 10 / (i - 10))
                .onErrorStop() //错误以后停止流。源头中断
                //.onErrorComplete()
                .subscribe(v -> System.out.println("v = " + v),
                        err -> System.out.println("err = " + err));

        Thread.sleep(100000);

    }

    class BusinessException extends RuntimeException {
        public BusinessException(String msg) {
            super(msg);
        }
    }

    Mono<Integer> hahaha(Throwable throwable) {
        if (throwable.getClass() == NullPointerException.class) {
            //...
        }

        return Mono.just(777);
    }

    @Test//超时与重试机制
    public void retryAndTimeout() throws InterruptedException {
        Flux.just(1)
                .delayElements(Duration.ofSeconds(3)) //一秒发送一个
                .log()
                .timeout(Duration.ofSeconds(2)) //规定超时时间为1秒
                .retry(3) //把流从头到尾重新请求一次
                .map(i -> i + "haha")
                .subscribe(System.out::println);


        Thread.sleep(10000);


    }

    @Test //sinks工具类
    public void sinks() throws InterruptedException {
        //Sinks允许我们快速建立通道，数据管道

//        Sinks.many() //发送Flux数据
//        Sinks.one() //发送Mono数据

//        Sinks.many().unicast(); //单播: 管道可以绑定单个订阅者【消费者】
//        Sinks.many().multicast(); //多播：管道可以绑定多个订阅者
//        Sinks.many().replay(); //重放：这个管道可以重放元素，后来的订阅者依然可以拿到之前的元素

        //从头消费还是从订阅的那一刻消费

        //背压队列
//        Sinks.Many<Object> many = Sinks.many()
//                //.unicast()
//                .multicast()
//                .onBackpressureBuffer();

        //默认订阅者，从订阅的那一刻开始接元素
        //重放队列
        Sinks.Many<Object> many = Sinks.many().replay().all();


        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                many.tryEmitNext("a-" + (i + 1));
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();

        many.asFlux().subscribe(v -> System.out.println("v1 = " + v));
        Thread.sleep(5000);
        many.asFlux().subscribe(v -> System.out.println("v2 = " + v));


        Thread.sleep(12000);
    }

    @Test //缓存 类似.replay().limit(N)
    public void cache() throws InterruptedException {
        Flux<Integer> cache = Flux.range(1, 10)
                .delayElements(Duration.ofSeconds(1))
                .cache(2)//可以缓存历史的2个数据 【如果不调缓存，默认缓存所有元素】
                ;


        cache.subscribe();//空订阅，使管道进行流动

        //因为带缓存了，所以前面三个元素会被缓存
        new Thread(() -> {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            cache.subscribe(v -> System.out.println("v2 = " + v));
        }).start();


        Thread.sleep(17000);

    }

    @Test //阻塞式API block
    public void block() {
        //把流转换成阻塞式API
        List<Integer> block = Flux.just(1, 2, 3)
                .map(i -> i + 10)
                .collectList().block();
        System.out.println("block = " + block);
    }

    @Test //并发流
    public void paralleFlux() throws InterruptedException {
        //百万数据 8个线程 每个线程处理100，进行分批处理一直处理结束
        Flux.range(1, 100)
                .buffer(10) //分发
                .publishOn(Schedulers.newParallel("xx")) //发布者指定线程池并行
                .parallel(8) //中间操作指定8个并行
                .runOn(Schedulers.newParallel("yy"))//中间并行 指定8个并发在那个线程池运行
                .log()
                //收集数据
                .flatMap(list -> Flux.fromIterable(list))//转换成可迭代即可
                .collectSortedList(Integer::compareTo)//将所有元素收集排序成一个集合
                .subscribe(v -> System.out.println("v = " + v));


        Thread.sleep(10000);


    }


    @Test //Context-API <- ThreadLocal在响应式编程中无法使用
    //响应式中，数据流共享数据，Context api ： Context : 读写 ContextView ：只读
    public void threadlocal() {
        /**
         * Context由下游传播给上游，上游可以拿到下游的最近一次数据
         * Context不可变
         */
        Flux.just(1, 2, 3)
                .transformDeferredContextual(
                        (flux, context) -> {
                            System.out.println("flux = " + flux);
                            System.out.println("context = " + context);
                            return flux.map(i -> i + " ===> " + context.get("prefix"));
                        }
                )
//                .map(i->{
//                    System.out.println("i = " + i);
//                    return i+"后缀";
//                })
                .contextWrite(Context.of("prefix", "哈哈")) //ThreadLocal共享了数据，上游的所有人都能看到
                .subscribe(v -> System.out.println("v = " + v));

        /**
         * 以前命令式编程
         * controller -> service -> dao
         *
         * 响应式编程
         * dao -> service -> controller 从下游反向传播
         */

    }


}
